package com.custom;


import com.custom.extra.MyRecordEmitter;
import com.custom.extra.MySplitReader;
import com.custom.source.ClinkSource;
import lombok.val;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.*;

/**
 * 具体的数据源
 */
public class MySource implements ClinkSource<Long, MySplit, MyEnumCheckpoint> {
    private final long from;
    private final long to;

    public MySource(long to) {
        this(0, to);
    }

    public MySource(long from, long to) {
        this.from = from;
        this.to = to;
    }

    @Override
    public Boundedness getBoundedness() {
        return Boundedness.BOUNDED;
    }

    @Override
    public SplitEnumerator<MySplit, MyEnumCheckpoint> createEnumerator(SplitEnumeratorContext<MySplit> enumContext) throws Exception {
        return new MySplitEnumerator(enumContext, from, to);
    }

    @Override
    public SplitEnumerator<MySplit, MyEnumCheckpoint> restoreEnumerator(SplitEnumeratorContext<MySplit> enumContext, MyEnumCheckpoint checkpoint) throws Exception {
        return new MySplitEnumerator(enumContext, checkpoint);
    }

    @Override
    public SourceReader<Long, MySplit> createReader(SourceReaderContext readerContext) throws Exception {
        val recordEmitter = new MyRecordEmitter();
        return new MyReader<>(MySplitReader::new, recordEmitter, readerContext.getConfiguration(), readerContext);
    }

    @Override
    public TypeInformation<Long> getProducedType() {
        return Types.LONG;
    }
}
